package com.atguigu.app;

import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * ClassName: Test001
 * Package: com.atguigu.app
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/19 9:43
 * @Version 1.0
 */
public class Test001 {
    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");


        //value={"common":{"sc":"1","ar":"33","uid":"1497","os":"Android 11.0","ch":"wandoujia","is_new":"0","md":"Xiaomi Mix2 ","mid":"mid_93","vc":"v2.1.111","ba":"Xiaomi","sid":"1d7455ef-4d64-46d4-b72e-9ae734f914ec"},"appVideo":{"play_sec":30,"position_sec":240,"video_id":"3716"},"ts":1684322854453}

        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_cart_info", "cart_add"), WatermarkStrategy.noWatermarks(), "kafka-source");
        kafkaDS.print();


        env.execute();

    }
}
